package javaVersion.learn.checkPoint;

import javaVersion.pojo.WordWithCount;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.runtime.state.StateBackend;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * @program: myFlink
 * @description: 单词统计，当遇到“error”字符串时，抛出异常，出现3次异常后，程序退出
 * @author: WincoMa
 * @create: 2020-07-22 18:00
 **/

public class FixDelayRestartStrategyDemo {
    public static void main(String[] args) {
        // TODO: 2020/7/22 初始化环境
        StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
        senv.enableCheckpointing(5000);
        StateBackend fsStateBackend = new FsStateBackend("hdfs://hadoop1/flinkCk/ck/");
        senv.setStateBackend(fsStateBackend);
        // TODO: 2020/7/22 设置Ck
        CheckpointConfig ck = senv.getCheckpointConfig();
        //保存checkpoint到hdfs
        ck.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        ck.setMaxConcurrentCheckpoints(1);
        // TODO: 2020/7/23 设置重启策略，默认重启策略是不停的重启。
        senv.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 5000));

        // TODO: 2020/7/22 读取socket数据源
        DataStreamSource<String> lines = senv.socketTextStream("hadoop1", 9999);

        // TODO: 2020/7/23 数据处理
        lines.flatMap(new FlatMapFunction<String, WordWithCount>() {
            @Override
            public void flatMap(String value, Collector<WordWithCount> out) throws Exception {
                String[] words = value.split("//s");
                for (String word : words) {
                    if (word.equals("error")) {
                        throw new RuntimeException("程序出错！");
                    }
                    out.collect(new WordWithCount(word, 1));
                }
            }
        })
                .keyBy(0)
                .timeWindow(Time.seconds(2), Time.seconds(1))
                .sum(1)
                .print();
        try {
            senv.execute();
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("运行出错，请检查");
        }
    }
}
